package com.freez.spark.core.operation;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import scala.Tuple2;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

/**
 * FREEDOM  2021 人生苦短，不妨一试
 * FileName: GetParaOpteration.java
 * Author: zcs
 * Date: 2021年-12月-07日 周二 18:02
 * Description: 获取元素类算子
 */
public class GetParamOperation {
    public static void main(String[] args) {
        SparkConf conf = new SparkConf();
        conf.setAppName("SparkCore-OperationDemo").setMaster("local");
        JavaSparkContext jsc = new JavaSparkContext(conf);
//数据
        List<Tuple2<String, Integer>> dataList = new ArrayList<>();
        dataList.add(new Tuple2<String, Integer>("2020-05-01", 11));
        dataList.add(new Tuple2<String, Integer>("2020-05-02", 4));
        dataList.add(new Tuple2<String, Integer>("2020-05-02", 7));
        dataList.add(new Tuple2<String, Integer>("2020-05-01", 4));
        dataList.add(new Tuple2<String, Integer>("2020-05-01", 10));
        JavaPairRDD<String, Integer> aRdd = jsc.parallelizePairs(dataList);
        JavaRDD<String> bRdd = jsc.parallelize(Arrays.asList("hello world", "hello hi"));
        /**
         * take和first 获取RDD中的某个元素
         * 其中take(1) == first
         */
        JavaRDD<String> oRdd = jsc.parallelize(Arrays.asList("hello world", "hello hi", "hello jack"));
        System.out.println("take>>>>>>>>" + oRdd.take(1));
        System.out.println("first>>>>>>>>" + oRdd.first());
        /**
         * lookup 查找RDD中指定key的元素value值，以List集合作为返回。
         */
        List<Integer> date = aRdd.lookup("2020-05-02");
        for (Integer item : date) {
            System.out.println(item);
        }
        /**
         * top 将RDD中的元素按照指定的排序规则(默认降序)，返回指定的前几个元素集合。
         */
        JavaRDD<Integer> intRDD = jsc.parallelize(Arrays.asList(3, 8, 2, 5, 7, 1, 4));
        List<Integer> top = intRDD.top(4);
        for (Integer result : top) {
            System.out.println(result);
        }
        /**
         * takeOrdered  与top相反，top()降序前几个，而takeOrdered()则是升序前几个,返回其List集合。
         */
        List<Integer> secList = intRDD.takeOrdered(4);
        for (Integer result : secList) {
            System.out.println(result);
        }
        /**
         * takeSample 用于对RDD中获得元素进行取样。</br>
         * 于sample()有所不同，首先sample()是以RDD形式返回抽样结果，而takeSample()是以List集合形式返回抽样结果。</br>
         * 其次sample()为转换算子，而takeSample()则是行动算子。
         */
        List<String> list = oRdd.takeSample(false, 2);
        for (String str: list){
            System.out.println(str);
        }

        jsc.close();

    }
}